/*
 * $AIST_Release: 0.9.0 $
 * Copyright 2011 Information Technology Research Institute, National
 * Institute of Advanced Industrial Science and Technology
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *    http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package aggr

import Aggregators._
import scala.collection.mutable.{Map => MMap}
import scala.collection.mutable.{Set => MSet}
import scala.collection.JavaConversions._

import com.google.protobuf.Descriptors._
import com.google.protobuf.Descriptors.FieldDescriptor.Type
import com.google.protobuf.Message
import com.google.protobuf.DynamicMessage

import util.JavaTypes._
import SerializedTypes._
import Util._

/**
 * "unique" aggregator.
 */
class Unique(N: Long) extends Aggregator {
  type Tuple = MMap[FieldDescriptor, Any]
  def emptyTuple: Tuple = MMap()

  def equal(lhs: Any, rhs: Any): Boolean = {
    (lhs, rhs) match {
      case (lhs: AnyRef, rhs: AnyRef) => lhs.equals(rhs)
    }
  }

  override def aggregate(data: TraversableOnce[(Any, Any)])(writer: (Any, Any) => Unit): Unit = {

    var m: MMap[Any, MSet[Any]] = MMap()

    data.foreach { case (k, v) =>
      val set = m.getOrElseUpdate(k, MSet())
      set += v
    }

    m.foreach { case (key, set) => 
      writer(key, set.size.asInstanceOf[Long])
    }
  }

  override def getValueType(): Option[SerializedType] = Some(Int64Type)
}

class UniqueFactory extends AggregatorFactory {

    override def createNew(
                  keyType: SerializedType,
                  valType: SerializedType,
                  args: List[Long]): Aggregator = {
      new Unique(args(0))
    }

    override def check(keyType: SerializedType,
              valType: SerializedType,
              args: List[SerializedType], 
              options: Map[String, SerializedType]): Option[List[String]]= {
      args.length == 1 && args(0) == Int64Type && options.isEmpty()

      val aggrName = "unique"

      val argsErrors = checkNumOfArgs(aggrName, args.length, 1) match {
        case Nil if (args(0) != Int64Type) => List("First arguments must be int type")
        case x => x
      }

      val optionErrors = checkOptionName(aggrName, options, List()) 

      (argsErrors ++ optionErrors) match {
        case Nil => None
        case x   => Some(x)
      }
    }
}
